package com.surfilter.massdata.spark.task;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;

import scala.Tuple2;

import com.act.sparkanalyz.service.impl.SparkService.OutQueueEntity;
import com.act.sparkanalyz.task.ISparkTask;
import com.surfilter.massdata.spark.bean.DnsRegionHour;
import com.surfilter.massdata.spark.bean.SA5601;
import com.surfilter.massdata.spark.bean.SA6601;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.DateUtil;
import com.surfilter.massdata.spark.util.URLUtil;

public class DomainAnalyzTask_SA5601_SA6601  implements ISparkTask{
	
	private static final Log log = LogFactory.getLog(DomainAnalyzTask_SA5601_SA6601.class);
	private static final long serialVersionUID = 1L;

	@SuppressWarnings("serial")
	@Override
	public List<OutQueueEntity> execute(Map<String, DataFrame> dataFrames, Map<String, String> commandMap) {
		log.info("==============DomainAnalyzTask_SA5601_SA6601 begin====================");
		List<OutQueueEntity> outList = new ArrayList<OutQueueEntity>();
		try{
			DataFrame dataSource = dataFrames.get("dip_frame");
			SQLContext sqlContext = dataSource.sqlContext();
			
			JavaRDD<Row> dipFilterRDD = dataSource.toJavaRDD();
			
			JavaPairRDD<String, Integer> domainCounts = getPairRDD(dipFilterRDD);
	        
           DataFrame domainCountFrame = getDomainCountFrame(sqlContext,dipFilterRDD);
	        
//	        long count = dipFilterRDD.mapToPair(new PairFunction<Row,String,Long>() {
//
//				@Override
//				public Tuple2<String,Long> call(Row row) throws Exception {
//					String domain = row.getAs("domain");
//					return new Tuple2<String,Long>(domain,1L);
//				}
//			}).reduceByKey(new Function2<Long, Long, Long>() {
//
//				@Override
//				public Long call(Long v1, Long v2) throws Exception {
//					return v1 + v2;
//				}
//			}).count();
//	        
	        long domainCount = domainCounts.count();
	        
	    	List<SA5601> list_SA5601 = new ArrayList<SA5601>();
	    	//List<SA6601> list_SA6601 = new ArrayList<SA6601>();
	    	
	        String dayStr = commandMap.get("-d");
	        Date date = DateUtil.getExecDate(DateUtil.getYesterday(), dayStr);
	        
	        if(domainCount > 0){
	        	SA5601 sa = new SA5601();
	     		sa.setYear(DateUtil.getCurrentYear(date));
	     		sa.setHalf_year(DateUtil.getHalfYear(date));
	     		sa.setQuarter(DateUtil.getQuarter(date));
	     		sa.setMonth(DateUtil.getCurrentMonth(date));
	     		sa.setWeek(DateUtil.getCurrentWeek(date));
	     		sa.setDay(DateUtil.getCurrentDay(date));
	     		sa.setBuss_type("WEBSITE_COUNT_SUM");
	     		sa.setBuss_value("1");
	     		sa.setCount_value(domainCount);
	     		sa.setSta_range(0);
	     		list_SA5601.add(sa);
	     		
//	     		SA6601 sa6601 = new SA6601();
//	     	    sa6601.setYear(DateUtil.getCurrentYear(date));
//	     	    sa6601.setHalf_year(DateUtil.getHalfYear(date));
//	     	    sa6601.setQuarter(DateUtil.getQuarter(date));
//	     	    sa6601.setMonth(DateUtil.getCurrentMonth(date));
//	     	    sa6601.setWeek(DateUtil.getCurrentWeek(date));
//	     	    sa6601.setDay(DateUtil.getCurrentDay(date));
//	     	    sa6601.setBuss_type("WEBSITE_COUNT_SUM");
//	     	    sa6601.setBuss_value("");
//	     	    /*List<Row> list = domainCountFrame.toJavaRDD().collect();
//	     	    if(list != null && list.size() > 0){
//	     	    	sa6601.setCount_value(list.get(0).getLong(0));
//	     	    }*/
//	     	    sa6601.setCount_value(count);
//	     	    sa6601.setSta_range(0);
//	     	    list_SA6601.add(sa6601);
	        }
	        	   
	    	JavaSparkContext ctx = new JavaSparkContext(dataSource.sqlContext().sparkContext());
	    	JavaRDD<SA5601> rdd = ctx.parallelize(list_SA5601);
	    	//JavaRDD<SA6601> rdd_SA6601 = ctx.parallelize(list_SA6601);
	    	DataFrame df = dataSource.sqlContext().createDataFrame(rdd, SA5601.class);
	    	//DataFrame df_SA6601 = dataSource.sqlContext().createDataFrame(rdd_SA6601, SA6601.class);
	    	
	    	//sqlContext.dropTempTable("domain_result_temp");
			outList.add(new OutQueueEntity("SA5601_TEMP", df));
			//outList.add(new OutQueueEntity("SA6601_TEMP", df_SA6601));
			
			CommonUtils.deleteTaskTableData("SA5601", date, "WEBSITE_COUNT_SUM", 0, "day");
			//CommonUtils.deleteTaskTableData("SA6601", date, "WEBSITE_COUNT_SUM", 0, "day");
		}catch(Exception e){
			if(log.isErrorEnabled()){
				log.error(e.getMessage(),e);
			}
		}
		return outList;
	}

	@SuppressWarnings("serial")
	private DataFrame getDomainCountFrame(SQLContext sqlContext,JavaRDD<Row> dipFilterRDD) {
		JavaRDD<DnsRegionHour> dnsRDD = dipFilterRDD.map(new Function<Row, DnsRegionHour>() {

			@Override
			public DnsRegionHour call(Row row){
				DnsRegionHour dns = new DnsRegionHour();
				String domain = row.getAs("domain");
				dns.setDomain(domain);
				return dns;
			}
		});
		
		sqlContext.createDataFrame(dnsRDD, DnsRegionHour.class).registerTempTable("domain_result_temp");
		DataFrame domainCountFrame = sqlContext.sql("select count(distinct domain) count from domain_result_temp");
		return domainCountFrame;
	}

	private JavaPairRDD<String, Integer> getPairRDD(JavaRDD<Row> dipFilterRDD) {
		JavaRDD<String> domainSuffix = dipFilterRDD.map(new Function<Row,String>(){
			
			private static final long serialVersionUID = 1L;
			
			@Override
			public String call(Row row) throws Exception {
				String domain = row.getAs("domain");
				String topDomain = "";
				try{
					 topDomain = URLUtil.getDomainName("http://" + domain);
				}catch(Exception e){
					if(log.isErrorEnabled()){
						log.error(e.getMessage(),e);
					}
				}
				return topDomain;
			}});
		
		JavaPairRDD<String,Integer> domainPair = domainSuffix.mapToPair(new PairFunction<String,String,Integer>(){
			
			private static final long serialVersionUID = 1L;
			
			@Override
			public Tuple2<String, Integer> call(String domain) throws Exception {
				Tuple2<String,Integer> tuple = new Tuple2<String,Integer>(domain,1);
				return tuple;
			}});
		
		JavaPairRDD<String,Integer> domainCounts = domainPair.reduceByKey(new Function2<Integer,Integer,Integer>(){
			
			private static final long serialVersionUID = 1L;
			
			@Override
			public Integer call(Integer x, Integer y)throws Exception {
				return x+y;
			}});
		
		return domainCounts;
	}
}
